package com.zx.learn.flink.apps.app1mysqltomysql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;

/**
 * MySQL 到MySQL
 */
@Slf4j
public class MySQLToMySQLMainSourceSimple {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.source

        DataStreamSource<SampleUser> sourceDS = env.addSource(new MySQLSource<SampleUser>() {
            @Override
            public void run(SourceContext<SampleUser> sourceContext) throws Exception {
                ResultSet resultSet = ps.executeQuery();
                while (resultSet.next()) {
                    SampleUser obj = new SampleUser();
                    obj.setId(resultSet.getInt("id"));
                    obj.setName(resultSet.getString("name"));
                    obj.setAge(resultSet.getInt("age"));
                    sourceContext.collect(obj);
                }
            }
            @Override
            Connection getConnection() {
                Connection con = null;
                try {
                    Class.forName("com.mysql.cj.jdbc.Driver");
                    //注意，替换成自己本地的 mysql 数据库地址和用户名、密码
                    con = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true&autoReconnect=true", "root", "root");
                } catch (Exception e) {
                    log.error("get connection has exception , msg = {}", e.getMessage());
                }
                return con;
            }

            @Override
            String getSql() {
                return "select * from sample_user limit 100";
            }

        });

        sourceDS.print();
        env.execute();
    }


}
